package com.zto.wk.webflux.kafka;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Properties;

/**
 * Created by wangkuan at 2019-02-22 10:10
 */
public class Consumer {
    private static Logger log = LoggerFactory.getLogger(Producer.class);

    private static final String TOPIC = "test_1";
    private static final String GROUP_ID = "test2";
    private static final String BOOTSTRAP_SERVERS = "kafka:9092";
    private static final KafkaConsumer<String, String> consumer;

    static {
        consumer = new KafkaConsumer<>(initConfig());
    }

    private static Properties initConfig() {
        Properties props = new Properties();
        props.put("bootstrap.servers", BOOTSTRAP_SERVERS);
        //每个消费者分配独立的组号
        props.put("group.id", GROUP_ID);
        //如果value合法，则自动提交偏移量
        props.put("enable.auto.commit", "true");
        //设置多久一次更新被消费消息的偏移量
        props.put("auto.commit.interval.ms", "1000");
        //设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息
        props.put("session.timeout.ms", "30000");
        //自动重置offset
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");

        return props;
    }

    public static void main(String[] args) {
        consumer.subscribe(Collections.singletonList(TOPIC));

        consumer.poll(100).forEach(e -> {
            System.out.println(e.key());
            System.out.println(e.value());
            System.out.println(e.offset());
            System.out.println();
        });
    }
}
